package com.kevin.netty.starter.client;

import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioChannelOption;
import io.netty.channel.socket.nio.NioSocketChannel;
import lombok.Getter;
import lombok.Setter;
import lombok.ToString;
import lombok.experimental.Accessors;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * 模拟客户端
 * 用于模拟会话数据传输
 */
@Slf4j
@Getter
@Setter
@ToString
@Accessors(chain = true)
public abstract class AbstractClient implements IClient {

    private String ip;

    private int port;

    private volatile boolean isConnected;

    private volatile boolean isClosed;

    private volatile boolean isMultiThread;

    private int connectNum;

    private int connectMaxNum = 3;

    private ExecutorService executor;

    private boolean reConnect;

    private Bootstrap bootstrap;

    private Channel channel;

    private NioEventLoopGroup workerGroup;

    public void connect(String ip, int port, ClientListener listener) {
        try {
            workerGroup = new NioEventLoopGroup();
            beforeBootstrap();
            bootstrap = new Bootstrap();
            bootstrap.group(workerGroup).channel(NioSocketChannel.class)
                    .remoteAddress(ip, port)
                    .option(ChannelOption.SO_KEEPALIVE, true)
                    .option(ChannelOption.TCP_NODELAY, true)
                    .option(ChannelOption.SO_REUSEADDR, true)
                    .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
                    .option(NioChannelOption.SO_SNDBUF, 1024 * 1024 * 1024)
                    .option(NioChannelOption.SO_RCVBUF, 1024 * 1024 * 1024)
                    .handler(getChannelInitializer());
            channel = bootstrap.connect().addListener((ChannelFutureListener) future -> {
                if (future.isSuccess()) {
                    log.warn("host : {} , port : {} , client connect succeed!", ip, port);
                    isConnected = true;
                    connectNum = 0;
                    if (listener != null) {
                        listener.onConnected();
                    }
                }
            }).sync().channel();
            afterBootstrap();
            channel.closeFuture().addListener((ChannelFutureListener) future -> {
                isConnected = false;
                log.warn("Client connection closed!");
                if (listener != null) {
                    listener.onClosed();
                }
            });
        } catch (Exception e) {
            throw new RuntimeException("Client connect error!", e);
        } finally {
            if (reConnect) {
                isConnected = false;
                workerGroup.shutdownGracefully();
                reConnect();
            }
        }
    }

    public void connect() {
        connect(null);
    }

    public void connect(ClientListener listener) {
        if (isMultiThread) {
            this.getExecutor().execute(() -> connect(ip, port, listener));
        } else {
            connect(ip, port, listener);
        }
    }

    public void close() {
        if (!isClosed && channel != null && channel.isOpen()) {
            isClosed = true;
        }
    }

    public void reConnect() {
        if (!isClosed && !isConnected && (connectNum++ < connectMaxNum)) {
            log.debug("Client reconnect num: {}", connectNum);
            try {
                Thread.sleep(10000);
                connect();
            } catch (Exception e) {
                log.error(e.getMessage(), e);
            }
        }
    }

    public abstract ChannelInitializer<SocketChannel> getChannelInitializer();

    public abstract void beforeBootstrap();

    public abstract void afterBootstrap();

    public synchronized void start() {
        connect();
    }

    public synchronized void stop() {
        workerGroup.shutdownGracefully();
    }

    public ExecutorService getExecutor() {
        return (null == this.executor) ? new ThreadPoolExecutor(
                Runtime.getRuntime().availableProcessors() * 2,
                Runtime.getRuntime().availableProcessors() * 2, 180, TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(500),
                new ThreadPoolExecutor.CallerRunsPolicy()) : this.executor;
    }

}
